package com.airlenet.dubbo.registry.mysql;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.support.FailbackRegistry;
import org.apache.dubbo.rpc.RpcException;

import javax.sql.DataSource;
import java.io.Serializable;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.*;

import static java.util.concurrent.Executors.newCachedThreadPool;
import static org.apache.dubbo.common.constants.CommonConstants.*;
import static org.apache.dubbo.common.constants.RegistryConstants.DYNAMIC_KEY;
import static org.apache.dubbo.registry.Constants.*;

public class MySQLRegistry extends FailbackRegistry {

    private static final Logger logger = LoggerFactory.getLogger(MySQLRegistry.class);

    private final int expirePeriod;
    private static final String TABLE_NAME = "`DUBBO_SERVICE`";
    private final ScheduledExecutorService expireExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("DubboRegistryExpireTimer", true));

    private final ScheduledFuture<?> expireFuture;
    private ExecutorService notifierExecutor = newCachedThreadPool(
            new NamedThreadFactory("dubbo-hazelcast-notifier", true));
    private final ConcurrentMap<URL, Notifier> notifiers = new ConcurrentHashMap<>();
    private final static String DEFAULT_ROOT = "dubbo";
    DataSource dataSource;
    private final String root;

    public MySQLRegistry(URL url, DataSource dataSource) {
        super(url);
        if (url.isAnyHost()) {
            throw new IllegalStateException("mysql address == null");
        }
        String group = url.getParameter(GROUP_KEY, DEFAULT_ROOT);
        this.root = group;
        this.dataSource = dataSource;

        this.expirePeriod = url.getParameter(SESSION_TIMEOUT_KEY, DEFAULT_SESSION_TIMEOUT);
        this.expireFuture = expireExecutor.scheduleWithFixedDelay(() -> {
            try {
                deferExpired(); // Extend the expiration time
            } catch (Throwable t) { // Defensive fault tolerance
                logger.error("Unexpected exception occur at defer expire time, cause: " + t.getMessage(), t);
            }
        }, expirePeriod / 2, expirePeriod / 2, TimeUnit.MILLISECONDS);
    }

    private void deferExpired() {
        for (URL url : new HashSet<>(getRegistered())) {
            if (url.getParameter(DYNAMIC_KEY, true)) {
                try (Connection connection = dataSource.getConnection()) {
                    PreparedStatement preparedStatement = connection.prepareStatement("UPDATE " + TABLE_NAME + "SET expired=?  WHERE  `group` =? AND side= ? AND service =? AND url= ?");
                    preparedStatement.setLong(1, System.currentTimeMillis() + expirePeriod);
                    preparedStatement.setString(2, url.getParameter(GROUP_KEY, root));
                    preparedStatement.setString(3, getSide(url));
                    preparedStatement.setString(4, url.getServiceInterface());
                    preparedStatement.setString(5, url.toFullString());
                    preparedStatement.executeUpdate();
                } catch (SQLException throwables) {
                    throw new RuntimeException(throwables);
                }
            }
        }
    }

    @Override
    public void register(URL url) {
//        if (isConsumerSide(url)) {
//            return;
//        }

        super.register(url);
    }

    @Override
    public void doRegister(URL url) {

        Long expire = (System.currentTimeMillis() + expirePeriod);

        try (Connection connection = dataSource.getConnection()) {
            PreparedStatement preparedStatement = connection.prepareStatement("INSERT INTO " + TABLE_NAME + " (`group`,side,service,created,expired,url) VALUES(?,?,?,?,?,?)");
            preparedStatement.setString(1, url.getParameter(GROUP_KEY, root));
            preparedStatement.setString(2, getSide(url));
            preparedStatement.setString(3, url.getServiceInterface());
            preparedStatement.setTimestamp(4, new Timestamp(System.currentTimeMillis()));
            preparedStatement.setLong(5, expire);
            preparedStatement.setString(6, url.toFullString());
            preparedStatement.executeUpdate();

        } catch (SQLException throwables) {
            throw new RuntimeException(throwables);
        }
//        hazelcastInstance.getMultiMap("dubbo-service").put(service, url);
//        String uuid = hazelcastInstance.getCluster().getLocalMember().getUuid();
//        hazelcastInstance.getMultiMap("dubbo-provider").put(uuid, new HazelcastService(service, url));
    }

    @Override
    public void unregister(URL url) {
//        if (isConsumerSide(url)) {
//            return;
//        }

        super.unregister(url);
    }

    @Override
    public void doUnregister(URL url) {
        String service = url.getServiceInterface();
        Long expire = (System.currentTimeMillis() + expirePeriod);

        try (Connection connection = dataSource.getConnection()) {
            url.toIdentityString();
            PreparedStatement preparedStatement = connection.prepareStatement("DELETE FROM " + TABLE_NAME + " WHERE  `group` =? AND side= ? AND service =? AND url= ?");
            preparedStatement.setString(1, url.getParameter(GROUP_KEY, root));
            preparedStatement.setString(2, getSide(url));
            preparedStatement.setString(3, url.getServiceInterface());
            preparedStatement.setString(4, url.toFullString());
            preparedStatement.executeUpdate();

        } catch (SQLException throwables) {
            throw new RuntimeException(throwables);
        }

    }

    @Override
    public void subscribe(URL url, NotifyListener listener) {
        if (isProviderSide(url)) {
            return;
        }

        super.subscribe(url, listener);
    }

    @Override
    public void doSubscribe(URL url, NotifyListener listener) {

        try (Connection connection = dataSource.getConnection()) {
            ResultSet resultSet = null;
            if (ANY_VALUE.equals(url.getServiceInterface())) {
                PreparedStatement preparedStatement = connection.prepareStatement("SELECT url FROM " + TABLE_NAME + " WHERE  `group` =? AND side= ? ");
                preparedStatement.setString(1, url.getParameter(GROUP_KEY, root));
                preparedStatement.setString(2, "provider");
                resultSet = preparedStatement.executeQuery();
            } else {
                PreparedStatement preparedStatement = connection.prepareStatement("SELECT url FROM " + TABLE_NAME + " WHERE  `group` =? AND side= ? AND service =?");
                preparedStatement.setString(1, url.getParameter(GROUP_KEY, root));
                preparedStatement.setString(2, "provider");
                preparedStatement.setString(3, url.getServiceInterface());
                resultSet = preparedStatement.executeQuery();
            }
            List<URL> urls = new ArrayList<>();
            while (resultSet.next()) {
                urls.add(URL.valueOf(resultSet.getString(1)));
            }
            notify(url, listener, urls);
            Notifier notifier = notifiers.computeIfAbsent(url, k -> new Notifier(url));
            notifierExecutor.submit(notifier);
        } catch (SQLException throwables) {
            throw new RuntimeException(throwables);
        }

    }

    @Override
    public void unsubscribe(URL url, NotifyListener listener) {
        if (isProviderSide(url)) {
            return;
        }

        super.unsubscribe(url, listener);
    }

    @Override
    public void doUnsubscribe(URL url, NotifyListener listener) {
        Notifier notifier = notifiers.remove(url);
        notifier.stop();
    }

    @Override
    public List<URL> lookup(URL url) {
        if (url == null) {
            throw new IllegalArgumentException("lookup url == null");
        }
        try (Connection connection = dataSource.getConnection()) {
            PreparedStatement preparedStatement = connection.prepareStatement("SELECT url FROM " + TABLE_NAME + " WHERE  `group` =? AND side= ? AND service =?");
            preparedStatement.setString(1, url.getParameter(GROUP_KEY, root));
            preparedStatement.setString(2, "provider");
            preparedStatement.setString(3, url.getServiceInterface());
            ResultSet resultSet = preparedStatement.executeQuery();
            List<URL> urls = new ArrayList<>();
            while (resultSet.next()) {
                urls.add(URL.valueOf(resultSet.getString(1)));
            }
            return urls;
        } catch (Throwable e) {
            throw new RpcException("Failed to lookup " + url + " from mysql " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

    @Override
    public boolean isAvailable() {
        return dataSource != null;
    }

    @Override
    public void destroy() {
        super.destroy();
        try {
            expireFuture.cancel(true);
        } catch (Throwable t) {
            logger.warn(t.getMessage(), t);
        }
        try {
            for (Notifier notifier : notifiers.values()) {
                notifier.stop();
            }
        } catch (Throwable t) {
            logger.warn(t.getMessage(), t);
        }
        dataSource = null;
    }

    private boolean isConsumerSide(URL url) {
        return url.getProtocol().equals(CONSUMER_PROTOCOL);
    }

    private String getSide(URL url) {
        return url.getParameter("side");
    }

    private boolean isProviderSide(URL url) {
        return url.getProtocol().equals(PROVIDER_PROTOCOL);
    }

    public void createTable() {
        try (Connection connection = dataSource.getConnection()) {
            PreparedStatement preparedStatement = connection.prepareStatement("CREATE TABLE  IF NOT EXISTS " + TABLE_NAME + " (\n" +
                    "  `id` bigint(20) NOT NULL AUTO_INCREMENT,\n" +
                    "  `group` varchar(255) COLLATE utf8_bin DEFAULT NULL ,\n" +
                    "  `side` varchar(255) COLLATE utf8_bin DEFAULT NULL COMMENT 'provider\\\\consumer',\n" +
                    "  `service` varchar(255) COLLATE utf8_bin DEFAULT NULL,\n" +
                    "  `created` datetime DEFAULT NULL,\n" +
                    "  `expired` bigint(20) DEFAULT NULL COMMENT '过期时间',\n" +
                    "  `url` text COLLATE utf8_bin,\n" +
                    "  PRIMARY KEY (`id`),\n" +
                    "    KEY (`group`,`side`,`service`)\n" +
                    ") ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;");
            preparedStatement.executeUpdate();

        } catch (SQLException throwables) {
            throw new RuntimeException(throwables);
        }
    }

    private class Notifier implements Runnable {
        private URL url;
        private boolean running;

        Notifier(URL url) {
            this.url = url;
            this.running = true;
        }

        @Override
        public void run() {
            while (this.running) {
                if (ANY_VALUE.equals(url.getServiceInterface())) {
                    processServices();
                } else {
                    processService();
                }
                try {
                    Thread.sleep(expirePeriod);
                } catch (InterruptedException e) {
                    // ignore
                }
            }
        }

        private void processService() {
            try (Connection connection = dataSource.getConnection()) {
                ResultSet resultSet = null;

                PreparedStatement preparedStatement = connection.prepareStatement("SELECT url FROM " + TABLE_NAME + " WHERE  `group` =? AND side= ? AND service =?");
                preparedStatement.setString(1, url.getParameter(GROUP_KEY, root));
                preparedStatement.setString(2, getSide(url));
                preparedStatement.setString(3, url.getServiceInterface());
                resultSet = preparedStatement.executeQuery();

                List<URL> urls = new ArrayList<>();
                while (resultSet.next()) {
                    urls.add(URL.valueOf(resultSet.getString(1)));
                }
                for (NotifyListener listener : getSubscribed().get(url)) {
                    doNotify(url, listener, urls);
                }
            } catch (SQLException throwables) {
                throw new RuntimeException(throwables);
            }

        }

        private void processServices() {
            try (Connection connection = dataSource.getConnection()) {
                ResultSet resultSet = null;

                PreparedStatement preparedStatement = connection.prepareStatement("SELECT url FROM " + TABLE_NAME + " WHERE  `group` =? AND side= ? ");
                preparedStatement.setString(1, url.getParameter(GROUP_KEY, root));
                preparedStatement.setString(2, getSide(url));
                resultSet = preparedStatement.executeQuery();

                List<URL> urls = new ArrayList<>();
                while (resultSet.next()) {
                    urls.add(URL.valueOf(resultSet.getString(1)));
                }
                for (NotifyListener listener : getSubscribed().get(url)) {
                    doNotify(url, listener, urls);
                }
            } catch (SQLException throwables) {
                throw new RuntimeException(throwables);
            }
        }

        void stop() {
            this.running = false;
        }
    }

}
